package io.wen.bd.s3m3.hbase;

import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

public class FriendsApplication {
    private static final String JAR_PATH = "/app/hbase/processor/friends-processor.jar";
    private static final String FRIENDS_TABLE = "t_friends";
    private static final String QUORUMS = "hadoop01,hadoop02,hadoop03";
    private static final String QUORUMS_PORT = "2181";
    private static final byte[] FRIENDS_FAMILY = "friends".getBytes();
    private static Connection connection;

    public static void main(String[] args) throws IOException {
        init();
        prepareData(args[0]);
//        deleteFriends();
    }

    /**
     * 删除uid1的好友uid2，检查uid2是否同样删除uid1
     */
    private static void deleteFriends() throws IOException {
        System.out.println("delete friends uid1 ...");
        final Table friendsTable = connection.getTable(TableName.valueOf(FRIENDS_TABLE));
        final Delete delete = new Delete("uid1".getBytes());
        delete.addColumn(FRIENDS_FAMILY, "uid2".getBytes());
        friendsTable.delete(delete);
        friendsTable.close();
    }

    /**
     * 将firiends.txt文件加载到t_friends表
     * @param dataFilePath 文件路径
     */
    private static void prepareData(String dataFilePath) throws IOException {
        System.out.println("prepare data ...");
        final Table friendsTable = connection.getTable(TableName.valueOf(FRIENDS_TABLE));
        List<Put> putList = new ArrayList<>();
        final List<String> lines = FileUtils.readLines(new File(dataFilePath));
        for (String line : lines) {
            final String[] split = line.split(":");
            final String[] friends = split[1].split(",");
            Put put = new Put(split[0].getBytes());
            for (String friend : friends) {
                put.addColumn(FRIENDS_FAMILY, friend.getBytes(), friend.getBytes());
            }
            putList.add(put);
        }
        friendsTable.put(putList);
        friendsTable.close();
    }

    /**
     * 初始化表，挂载协处理器
     * @throws IOException
     */
    private static void init() throws IOException {
        final Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", QUORUMS);
        conf.set("hbase.zookeeper.property.clientPort", QUORUMS_PORT);
        connection = ConnectionFactory.createConnection(conf);
        Admin admin = connection.getAdmin();

        final TableName tableName = TableName.valueOf(FRIENDS_TABLE);
        if (admin.tableExists(tableName)) {
            if (!admin.isTableDisabled(tableName)) {
                admin.disableTable(tableName);
            }
            System.out.println("delete table t_friends ...");
            admin.deleteTable(tableName);
        }

        // 建表，指定列族和协处理器
        final HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
        tableDescriptor.addFamily(new HColumnDescriptor(FRIENDS_FAMILY));
        tableDescriptor.addCoprocessor(FriendDeleteProcessor.class.getName(), new Path(JAR_PATH), 1001, new HashMap<>());
        admin.createTable(tableDescriptor);

        admin.close();
    }

}
